/*
 *
 * Copyright (c) 2020 The Raptor Authors. All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

#include "src/windows/container_impl.h"
#include <string.h>

#include "raptor-lite/impl/endpoint.h"
#include "raptor-lite/utils/log.h"
#include "raptor-lite/utils/mpscq.h"
#include "raptor-lite/utils/time.h"

#include "src/common/connection_id.h"
#include "src/common/endpoint_impl.h"
#include "src/common/resolve_address.h"
#include "src/common/socket_util.h"

#include "src/windows/connection.h"
#include "src/windows/iocp_thread.h"

namespace raptor {

enum ContainerMessageType { kRecvAMessage, kCloseEvent, kHeartbeatEvent };

struct ContainerMessageNode {
    MultiProducerSingleConsumerQueue::Node node;
    ContainerMessageType type;
    Endpoint ep;
    Slice slice;
    raptor_error desc;
    ContainerMessageNode(Endpoint e)
        : ep(e) {}

    ContainerMessageNode(Endpoint e, raptor_error ev)
        : ep(e)
        , desc(ev) {}
};

ContainerImpl::ContainerImpl(Option *option)
    : shutdown_(true)
    , magic_number_(0) {
    memcpy(&option_, option, sizeof(option_));
}

ContainerImpl::~ContainerImpl() { Shutdown(); }

raptor_error ContainerImpl::Init(PollingThread *pt) {
    if (!shutdown_) return RAPTOR_ERROR_FROM_STATIC_STRING("tcp server already running");

    raptor_error e = RAPTOR_ERROR_NONE;

    // init mq thread
    bool success = false;
    thd_ = Thread("ContainerImpl",
                  std::bind(&ContainerImpl::MessageQueueThread, this, std::placeholders::_1),
                  nullptr, &success);
    if (!success) {
        log_error("ContainerImpl: Failed to create mq thread");
        return RAPTOR_ERROR_FROM_STATIC_STRING("ContainerImpl: Failed to create mq thread");
    }

    // init timer thread
    if (option_.check_receiving_interval || option_.heartbeat_handler) {
        timer_thread_ = std::make_shared<Timer>(this);
        e = timer_thread_->Init();
        if (e != RAPTOR_ERROR_NONE) {
            log_error("ContainerImpl: Failed to init timer thread, %s", e->ToString().c_str());
            return e;
        }
    } else {
        log_debug("ContainerImpl: Skip timeout timer creation");
    }

    int init_count = option_.init_container_size / option_.recv_send_threads;
    int max_count = option_.max_container_size / option_.recv_send_threads;
    int64_t now = GetCurrentMilliseconds();
    magic_number_ = static_cast<uint16_t>(((now / 1000) >> 16) & 0xffff);

    for (int i = 0; i < option_.recv_send_threads; i++) {
        MiniContainer *mc = new MiniContainer(static_cast<uint16_t>(i));

        e = mc->Init(init_count, max_count);
        if (e != RAPTOR_ERROR_NONE) {
            log_error("ContainerImpl: Failed to init MiniContainer, %s", e->ToString().c_str());
            delete mc;
            return e;
        }

        mc->SetOption(&option_);
        mc->SetMagic(magic_number_);
        mgr_.push_back(mc);
    }

    log_debug("ContainerImpl: Initialization completed");
    polling_thread_ = pt;

    shutdown_ = false;
    count_.Store(0);
    return RAPTOR_ERROR_NONE;
}

raptor_error ContainerImpl::Start() {
    if (shutdown_) {
        return RAPTOR_ERROR_FROM_STATIC_STRING("ContainerImpl has not been initialized");
    }

    thd_.Start();

    if (timer_thread_) {
        timer_thread_->Start();
    }

    for (size_t i = 0; i < mgr_.size(); i++) {
        mgr_[i]->Start();
    }

    if (option_.check_receiving_interval) {
        timer_thread_->SetTimer(internal::TID_TIMEOUT_CHECKING, 0, 3000);
    }

    return RAPTOR_ERROR_NONE;
}

void ContainerImpl::Shutdown() {
    if (!shutdown_) {
        log_warn("ContainerImpl: prepare to shutdown");

        if (timer_thread_) {
            timer_thread_->Shutdown();
        }

        for (size_t i = 0; i < mgr_.size(); i++) {
            mgr_[i]->Shutdown();
            delete mgr_[i];
        }

        mgr_.clear();

        shutdown_ = true;
        cv_.Signal();
        thd_.Join();

        // clear message queue
        bool empty = true;
        do {
            auto n = mpscq_.PopAndCheckEnd(&empty);
            auto msg = reinterpret_cast<ContainerMessageNode *>(n);
            if (msg != nullptr) {
                delete msg;
            }
        } while (!empty);

        log_warn("ContainerImpl: shutdown");
    }
}

bool ContainerImpl::SendMsg(uint32_t connection_id, const Slice &s) {
    if (!internal::VerifyConnectionId(connection_id, magic_number_)) {
        return false;
    }

    uint16_t id = internal::GetContainerId(connection_id);
    if (id >= mgr_.size()) {
        return false;
    }

    uint32_t index = internal::GetConnectionIndex(connection_id);
    mgr_[id]->SendToEndpoint(index, s);
    return true;
}

void ContainerImpl::CloseEndpoint(Endpoint ep, bool notify) {
    auto connection_id = ep.ConnectionId();

    if (!internal::VerifyConnectionId(connection_id, magic_number_)) {
        return;
    }

    uint16_t id = internal::GetContainerId(connection_id);
    if (id >= mgr_.size()) {
        return;
    }

    uint32_t index = internal::GetConnectionIndex(connection_id);
    mgr_[id]->CloseEndpoint(index, notify);
}

MiniContainer *ContainerImpl::FindMiniContainer() {
    if (mgr_.empty()) return nullptr;
    if (mgr_.size() == 1) return mgr_[0];

    int32_t count = INT32_MAX;
    int32_t index = 0;

    for (int32_t i = 0; i < static_cast<int32_t>(mgr_.size()); i++) {
        auto c = mgr_[i]->Count();
        if (c == 0) {
            return mgr_[i];
        }
        if (c < count) {
            index = i;
        }
    }
    return mgr_[index];
}

raptor_error ContainerImpl::AttachEndpoint(Endpoint ep) {
    auto mc = FindMiniContainer();
    if (mc == nullptr) {
        return RAPTOR_ERROR_FROM_FORMAT("ContainerImpl can't find MiniContainer, mgr : %u",
                                        mgr_.size());
    }
    std::shared_ptr<Connection> conn = mc->InsertEndpoint(ep);
    if (!conn) {
        return RAPTOR_ERROR_FROM_FORMAT("The number of connections has exceeded the limit : %u",
                                        option_.max_container_size);
    }

    ep.impl_->SetContainer(this);

    conn->SetPollingThread(polling_thread_);
    conn->SetProtocol(option_.proto_handler);
    conn->SetTransferService(this, this);

    if (timer_thread_ && option_.heartbeat_handler) {
        uint32_t timer_id = mc->Id();
        uint32_t index = conn->Index();

        timer_id = (timer_id << internal::kIndexOffset) | index;
        uint32_t delay = static_cast<uint32_t>(option_.heartbeat_handler->GetHeartbeatInterval());
        timer_thread_->SetTimer(internal::TID_HEARTBEAT, timer_id, delay);
    }
    return RAPTOR_ERROR_NONE;
}

// Receiver implement (iocp / epoll event)

void ContainerImpl::OnEventProcess(EventDetail *detail) {
    if (shutdown_ || !detail) return;

    auto connection_id = detail->overlaped->user_id;

    if (!internal::VerifyConnectionId(connection_id, magic_number_)) {
        return;
    }

    // replace the value of ptr, old value is socket_fd
#if defined(__x86_64__) || defined(_WIN64)
    detail->ptr = reinterpret_cast<void *>(static_cast<intptr_t>(connection_id));
#else
    detail->ptr = reinterpret_cast<void *>(connection_id);
#endif
    uint16_t id = internal::GetContainerId(connection_id);
    if (id >= mgr_.size()) {
        return;
    }
    mgr_[id]->EventHandler(detail);
}

// ServiceInterface implement

void ContainerImpl::OnDataReceived(Endpoint ep, const Slice &s) {
    if (shutdown_) return;
    auto msg = new ContainerMessageNode(ep);
    msg->slice = s;
    msg->type = ContainerMessageType::kRecvAMessage;
    mpscq_.push(&msg->node);
    count_.FetchAdd(1, MemoryOrder::RELAXED);
    cv_.Signal();
}

void ContainerImpl::OnClosed(Endpoint ep, raptor_error desc) {
    if (shutdown_) return;
    if (!option_.closed_handler) return;

    auto msg = new ContainerMessageNode(ep, desc);
    msg->type = ContainerMessageType::kCloseEvent;
    mpscq_.push(&msg->node);
    count_.FetchAdd(1, MemoryOrder::RELAXED);
    cv_.Signal();
}

void ContainerImpl::OnHeartBeat(Endpoint ep) {
    if (shutdown_) return;
    if (!option_.heartbeat_handler) return;

    auto msg = new ContainerMessageNode(ep);
    msg->type = ContainerMessageType::kHeartbeatEvent;
    mpscq_.push(&msg->node);
    count_.FetchAdd(1, MemoryOrder::RELAXED);
    cv_.Signal();

    // set new timer for heartbeat
    auto connection_id = ep.ConnectionId();
    uint32_t timer_id = internal::GetContainerId(connection_id);
    uint32_t index = internal::GetConnectionIndex(connection_id);

    timer_id = (timer_id << internal::kIndexOffset) | index;
    uint32_t delay = static_cast<uint32_t>(option_.heartbeat_handler->GetHeartbeatInterval());
    timer_thread_->SetTimer(internal::TID_HEARTBEAT, timer_id, delay);
}

void ContainerImpl::MessageQueueThread(void *) {
    while (!shutdown_) {
        if (count_.Load() < 1) {
            AutoMutex g(&cmtx_);
            cv_.Wait(&cmtx_, 1000);
            continue;
        }
        auto n = mpscq_.pop();
        auto msg = reinterpret_cast<struct ContainerMessageNode *>(n);
        if (msg != nullptr) {
            count_.FetchSub(1, MemoryOrder::RELAXED);
            this->Dispatch(msg);
            delete msg;
        }
    }
}

void ContainerImpl::Dispatch(struct ContainerMessageNode *msg) {
    switch (msg->type) {
    case ContainerMessageType::kRecvAMessage:
        option_.msg_handler->OnMessage(msg->ep, msg->slice);
        break;
    case ContainerMessageType::kCloseEvent:
        option_.closed_handler->OnClosed(msg->ep, msg->desc);
        break;
    case ContainerMessageType::kHeartbeatEvent:
        option_.heartbeat_handler->OnHeartbeat(msg->ep);
        break;
    default:
        log_error("ContainerImpl: Dispatch found an unknow message type %d",
                  static_cast<int>(msg->type));
        break;
    }
}

void ContainerImpl::OnTimer(uint32_t tid1, uint32_t tid2) {

    if (tid1 == internal::TID_TIMEOUT_CHECKING) {
        for (size_t i = 0; i < mgr_.size(); i++) {
            mgr_[i]->TimeoutChecking();
        }
        timer_thread_->SetTimer(internal::TID_TIMEOUT_CHECKING, 0, 3000);
        return;
    }

    if (tid1 == internal::TID_HEARTBEAT) {
        uint32_t id = tid2 >> internal::kIndexOffset;
        uint32_t index = tid2 & internal::kIndexBitsMask;
        if (id < mgr_.size()) {
            mgr_[id]->HeartBeatHandler(index);
        }
    }
}

} // namespace raptor
